package com.rrd.dw.utils;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionOutputStream;
import org.apache.hadoop.util.ReflectionUtils;

public class MergeHdfsFile {

	public static boolean mergeFile(String srcDir,String desDir)throws Exception{
		return mergeFile(srcDir,desDir,"org.apache.hadoop.io.compress.GzipCodec");
	}
	public static boolean mergeFile(String srcDir,String desDir,String codecClass)throws Exception{
		Configuration conf =new Configuration();
		CompressionCodec codec = (CompressionCodec)ReflectionUtils.newInstance(Class.forName(codecClass), conf);
		boolean bool= true;
		Path srcPath = new Path(srcDir);
		Path destPath = new Path(desDir);
		 CompressionOutputStream out =null;
		 FSDataInputStream in = null;
		try{
		FileSystem fs =FileSystem.get(conf);
		if(!fs.isDirectory(srcPath) ){
			System.out.println("目录不存在："+srcPath);
			return bool;
		}
		//判断目录是否存在，不存在创建
		if(!fs.exists(destPath) || !fs.isDirectory(destPath)){
			fs.mkdirs(destPath);
		}
		System.out.println("输出目录："+destPath);
		String outFile = desDir+"/"+System.currentTimeMillis()+".tar.gz";
		FSDataOutputStream outputStream = fs.create(new Path(outFile));
		
		out = codec.createOutputStream(outputStream);
		FileStatus[] list = fs.listStatus(srcPath);
		  for (FileStatus f : list) {
			  
			  if(f.getPath().getName().equals("_SUCCESS") || fs.isDirectory(f.getPath())){
				  System.out.println("过滤文件："+f.getPath());
				  continue;
			  }
			  System.out.println("读取文件："+f.getPath());
			  in = fs.open(f.getPath());
//	            log.info("name: "+f.getPath()+", folder: "+f.isDir()+", size: "+f.getLen()+"\n" );
			  IOUtils.copyBytes(in, out, conf,false);
			  
			  IOUtils.closeStream(in);
	        }
		  
		}finally{
			System.out.println("关闭输出流！！！");
			IOUtils.closeStream(out);
		}
		return bool;
	}
	public static void main(String[] args) {
		if(args==null || args.length<2){
			System.out.println("请输入参数：输入目录，输出目录");
			System.exit(1);
		}
		try{
			mergeFile(args[0], args[1]);
		}catch(Exception e){
			e.printStackTrace();
			System.exit(1);
		}
		
	}
}
